package com.afdor.rws.event.driven.support.channel.consumer;

import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.event.driven.api.EventBus;
import com.afdor.rws.event.driven.api.EventHolder;
import com.afdor.rws.event.driven.api.Serialization;
import com.afdor.rws.event.driven.api.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.api.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.api.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.support.channel.consumer.handler.ProxyEventHandler;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;

/**
 * 消费者抽象实现
 *
 * @author 悭梵
 * @date Created in 2018-12-27 9:34
 */
@Slf4j
public abstract class AbstractEventConsumer<EH extends EventHolder, CC extends EventConsumerConfig> implements EventConsumer<EH, CC> {
    /**
     * 线程运行状态
     */
    private AtomicBoolean running = new AtomicBoolean(Boolean.FALSE);
    /**
     * 消费者的名称
     */
    @Getter
    @NonNull
    private String consumerName;
    /**
     * 对应事件源
     */
    @Getter
    @NonNull
    private String eventSourceName;
    /**
     * 配置信息
     */
    @Getter
    @NonNull
    private CC config;
    /**
     * 所属事件总线
     */
    @Getter
    @NonNull
    private EventBus eventBus;
    /**
     * 事件处理
     */
    @Getter
    @NonNull
    private EventHandler eventHandler;
    /**
     * 消费失败重试策略
     */
    @Getter
    @NonNull
    private EventRetryStrategy eventRetryStrategy;
    /**
     * 序列化方式
     */
    @Getter
    @NonNull
    private Serialization serialization;

    @Getter
    private Map<String, String> filterTagMap;

    @Override
    public void start() {
        checkup(Boolean.FALSE);
        doStart();
        this.running.compareAndSet(Boolean.FALSE, Boolean.TRUE);
    }

    @Override
    public synchronized <T extends EventConsumer> T setConfig(@NonNull CC config) {
        if (StringUtils.isBlank(config.getConsumerName())) {
            throw new NullPointerException("消费者名称为空");
        }
        if (StringUtils.isBlank(config.getEventSourceName())) {
            throw new NullPointerException("事件源名称为空");
        }
        if (StringUtils.isBlank(config.getSerializationStrategyName())) {
            throw new NullPointerException("事件序列化策略名称为空");
        }

        this.config = config;
        this.consumerName = this.config.getConsumerName();
        this.eventSourceName = this.config.getEventSourceName();
        this.serialization = ExtensionLoader.getInstance(Serialization.class)
                .getExtensionOrDefault(this.config.getSerializationStrategyName());

        if (this.serialization == null) {
            throw new NullPointerException(this.config.getSerializationStrategyName() + "对应事件序列化策略不存在");
        }

        if (!(this.config.getFilterTagMap() == null || this.config.getFilterTagMap().isEmpty())) {
            this.filterTagMap = new HashMap<>(this.config.getFilterTagMap());
        }

        if (StringUtils.isBlank(config.getEventRetryStrategyName())) {
            this.eventRetryStrategy = ExtensionLoader.getInstance(EventRetryStrategy.class)
                    .newExtension(this.config.getEventRetryStrategyName());
        }
        return (T) this;
    }

    @Override
    public synchronized <T extends EventConsumer> T setEventBus(EventBus eventBus) {
        this.eventBus = eventBus;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setEventHandler(@NonNull EventHandler<EH> eventHandler) {
        this.eventHandler = new ProxyEventHandler(this, eventHandler);
        return (T) this;
    }


    @Override
    public <T extends EventConsumer> T setEventRetryStrategy(EventRetryStrategy eventRetryStrategy) {
        this.eventRetryStrategy = eventRetryStrategy;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T addFilterTag(String tagLabel, String value) {
        if (this.filterTagMap == null) {
            this.filterTagMap = new HashMap<>(8);
        }
        this.filterTagMap.put(tagLabel, value);
        return (T) this;
    }

    @Override
    public void stop() {
        this.running.compareAndSet(Boolean.TRUE, Boolean.FALSE);
        doStop();
    }

    /**
     * 启动消费者相关业务处理
     */
    protected void doStart() {

    }

    /**
     * 停止消费者相关业务处理
     */
    protected void doStop() {

    }

    protected boolean tagFilter(@NonNull Event event) {
        return Optional.ofNullable(getFilterTagMap())
                .orElse(Collections.emptyMap())
                .entrySet()
                .parallelStream()
                .filter(Objects::nonNull)
                .allMatch(entry -> {
                    String tagLabel = entry.getKey();
                    String filterValue = entry.getValue();
                    if (StringUtils.isBlank(filterValue)) {
                        filterValue = Event.EVENT_TAG_FILTER_ALL;
                    }
                    /* 匹配指定标签所有值时，要求标签一定要存在 */
                    if (Objects.equals(Event.EVENT_TAG_FILTER_ALL, filterValue)) {
                        return event.getTagMap().containsKey(tagLabel);
                    }
                    /* 获取指定标签值 */
                    String value = (String) event.getTagMap().get(tagLabel);
                    /* 为空 */
                    if (StringUtils.isBlank(value)) {
                        return Boolean.FALSE;
                    }
                    /* 相等 */
                    if (StringUtils.equals(filterValue, value)) {
                        return Boolean.TRUE;
                    }
                    try {
                        /* 正则匹配，可考虑正则表达式池缓存 */
                        Pattern pattern = Pattern.compile(filterValue);
                        return pattern.matcher(value).matches();
                    } catch (Exception e) {
                        log.warn("正则匹配过滤失败", e);
                        return Boolean.FALSE;
                    }
                });
    }

    /**
     * 验证是否启动
     */
    protected void checkup() {
        checkup(Boolean.TRUE);
    }

    /**
     * 验证是否启动
     */
    protected void checkup(boolean vaildRunning) {
        if (vaildRunning) {
            if (!isRunning()) {
                throw new RuntimeException("消费者未启动");
            }
        } else {
            if (Objects.isNull(getConfig())) {
                throw new NullPointerException("配置信息为空");
            }
            if (Objects.isNull(getEventHandler())) {
                throw new NullPointerException("处理器为空");
            }
        }
    }

    /**
     * 是否处于运行状态
     *
     * @return
     */
    protected boolean isRunning() {
        return this.running.get();
    }

}
